package com.qf.reliable;

import com.qf.util.RabbitMQUtil;
import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;

/**
 * @author zjw
 * @description 确保消息可靠性
 * @date 2021/11/9 9:53
 */
public class Publisher {

    @Test
    public void publish() throws Exception {
        //1. 获取连接对象
        Connection connection = RabbitMQUtil.getConnection();

        //2. 构建Channel
        Channel channel = connection.openChannel().get();

        String msg = "aaabbccc";
        //基于Channel开启Confirm机制
        channel.confirmSelect();
        // 确保消息发送到Exchange
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("消息成功的发送到Exchange");
            }
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("消息没有发送到Exchange，将消息存储到数据库、重新发送：" + msg);
            }
        });
        // 添加Return的回调
        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消息没有路由到Queue，exchange：" + exchange + "，routingKey：" + routingKey + "，msg：" + new String(body));
            }
        });
        // 构建消息的Properties属性
        AMQP.BasicProperties props = new AMQP.BasicProperties().builder()
                .deliveryMode(2)
                .build();
        //3. 发送消息到Exchange，发送消息时，设置消息的properties属性
        channel.basicPublish("","hello_queue",true,props,msg.getBytes());


        System.in.read();
    }

}
